#include "MLInterStream.h"
#include "MLClustering_KMeans.h"
#include "MLTools_Stat.h"

///////////////////////////////////////////////////////////////////////////
// CMLInterStreamCluster
CMLInterStreamCluster::CMLInterStreamCluster(CMLInterStream *pStream, int nID)
{
	m_pStream  = pStream;
	m_pDistFunc = pStream->GetDistanceFunc();
	m_vLS.zero();
	m_vSS.zero();
	m_TimeLS = 0.0;
	m_nID = nID;
}

CMLInterStreamCluster::~CMLInterStreamCluster(void)
{
}

int CMLInterStreamCluster::Create(const CMLDataSet& dataSet)
{
	int i;
	if (dataSet.IsEmpty())
	{
		ML_ASSERT(false);
		return -1;
	}

	// Calculate the security radius and average movement
	SlidingWindowElem_ elem;
	m_SlidingWindow.clear();
	m_vLS.zero();
	m_vSS.zero();
	m_TimeLS = 0;
	int dataSize = dataSet.Count();
	for (i=0; i<dataSize; i++)
	{
		const CMLDataPoint& dataPoint = dataSet.Get(i);
		m_vLS.add(dataPoint);
		m_vSS.add(dataPoint^2);
		// Push into the slide window
		elem.dataPoint = dataSet.Get(i);
		elem.nTime = m_pStream->GetTimeNow();
		m_TimeLS += elem.nTime;
		m_SlidingWindow.push_back(elem);
	}

	return 0;
}

int CMLInterStreamCluster::Create(const CMLDataPoint &dataPoint, int nTime)
{
	// Calculate the security radius and average movement
	SlidingWindowElem_ elem;
	m_SlidingWindow.clear();
	m_vLS.zero();
	m_vSS.zero();
	m_TimeLS = 0;
	Absorb(dataPoint, nTime);

	return 0;
}

int CMLInterStreamCluster::Absorb(const CMLDataPoint &dataPoint, int nTime)
{
	// Push the new data point and movement
	SlidingWindowElem_ newElem;
	newElem.dataPoint  = dataPoint;
	newElem.nTime  = nTime;
	m_vLS.add(newElem.dataPoint);
	m_vSS.add(newElem.dataPoint^2);
	m_TimeLS += nTime;

	// Push the new element
	m_SlidingWindow.push_back(newElem);
	return 0;
}

int CMLInterStreamCluster::Popout(int nTimeDeadline)
{
	while(m_SlidingWindow.size() > 0)
	{
		const SlidingWindowElem_& frontElem = m_SlidingWindow.front();
		if (frontElem.nTime < nTimeDeadline)
		{
			m_vLS.sub(frontElem.dataPoint);
			m_vSS.sub(frontElem.dataPoint^2);
			m_TimeLS -= frontElem.nTime;
			// Pop out
			m_SlidingWindow.pop_front();
		}
		else
		{
			break;
		}
	}
	return 0;
}

int CMLInterStreamCluster::Reset(void)
{
	return 0;
}

bool CMLInterStreamCluster::TriggerDieOutEvent(void)
{	
	if (GetSWLength() < 1)
		return true;

	return false;
}

bool CMLInterStreamCluster::TriggerSplitEvent(void)
{
	if (m_pStream->GetMaxRadius() < 0)
		return false;

	return GetRadius() > m_pStream->GetMaxRadius();
}

int CMLInterStreamCluster::ExecuteSplit(CMLInterStreamCluster* pNewCluster)
{
	CMLDataPoint center = GetCenter();

	// Find the farest data point
	double dMaxDist = 0.0;
	CMLDataPoint FarestPoint;
	std::list<SlidingWindowElem_>::const_iterator Iter;
	std::list<SlidingWindowElem_>::const_iterator PreIter;
	for (Iter=m_SlidingWindow.begin(); Iter!=m_SlidingWindow.end(); Iter++)
	{
		double dDist = m_pDistFunc->Distance(center, (*Iter).dataPoint);
		if (dDist > dMaxDist)
		{
			dMaxDist = dDist;
			FarestPoint = (*Iter).dataPoint;
		}
	}

	// Split the dataset
	Iter=m_SlidingWindow.begin();
	while(Iter != m_SlidingWindow.end())
	{
		const CMLDataPoint& dataPoint = (*Iter).dataPoint;
		double dDist1 = m_pDistFunc->Distance(dataPoint, FarestPoint);
		double dDist2 = m_pDistFunc->Distance(dataPoint, center);
		if (dDist1 < dDist2)
		{
			CMLDataPoint squareDataPoint = dataPoint^2;
			pNewCluster->m_vLS.add(dataPoint);
			pNewCluster->m_vSS.add(squareDataPoint);
			this->m_vLS.sub(dataPoint);
			this->m_vSS.sub(squareDataPoint);

			pNewCluster->m_SlidingWindow.push_back(*Iter);
			PreIter = Iter;
			Iter++;
			m_SlidingWindow.erase(PreIter);
		}
		else
		{
			Iter++;
		}
	}

	pNewCluster->Reset();
	this->Reset();

	return 0;
}

int CMLInterStreamCluster::ExecuteThrowOut(void)
{
	// Split throw part
	CMLInterStreamCluster* pThrowPart = new CMLInterStreamCluster(m_pStream, -1);
	ExecuteSplit(pThrowPart);

	// Die-out the throw part
	pThrowPart->ExecuteDieOut();

	delete pThrowPart;
	return 0;
}

int CMLInterStreamCluster::ExecuteDieOut(void)
{
	int i;
	std::list<SlidingWindowElem_>::const_iterator Iter;
	std::vector<CMLDataPoint> vCenters;

	// Get all other centers of clusters
	vCenters.resize(m_pStream->GetClusters().size());
	for (i=0; i<(int)m_pStream->GetClusters().size(); i++)
	{
		CMLInterStreamCluster* pOtherCluster = m_pStream->GetClusters()[i];
		if (pOtherCluster == NULL)
			continue;
		if (this == pOtherCluster)
			continue;
		vCenters[i] = pOtherCluster->GetCenter();
	}
	// Redistribute data points to other clusters
	for(Iter=m_SlidingWindow.begin(); Iter!=m_SlidingWindow.end(); Iter++)
	{
		double dMinDist = DBL_MAX;
		CMLInterStreamCluster* pNearestCluster = NULL;
		for (i=0; i<(int)m_pStream->GetClusters().size(); i++)
		{
			CMLInterStreamCluster* pOtherCluster = m_pStream->GetClusters()[i];
			if (pOtherCluster == NULL)
				continue;
			if (this == pOtherCluster)
				continue;
			double dDist = m_pDistFunc->Distance((*Iter).dataPoint, vCenters[i]);
			if (dDist < dMinDist)
			{
				pNearestCluster = pOtherCluster;
				dMinDist = dDist;
			}
		}
		// Absorb
		pNearestCluster->Absorb((*Iter).dataPoint, (*Iter).nTime);
	}

	return 0;
}

int CMLInterStreamCluster::ExecuteMerge(CMLInterStreamCluster *pNearCluster)
{
	std::list<SlidingWindowElem_>::const_iterator Iter = pNearCluster->m_SlidingWindow.begin();
	for (; Iter != pNearCluster->m_SlidingWindow.end(); Iter++)
	{
		Absorb((*Iter).dataPoint, (*Iter).nTime);
	}
	Reset();
	return 0;
}


///////////////////////////////////////////////////////////////////////////
// CMLInterStream
CMLInterStream::CMLInterStream(void)
{
	m_pDistFunc = new CMLWeightedEuclideanDistance(CMLDataPoint(1));
	m_nInitNumber = 16;
	m_nNumInitClusters = 4;
	m_nDataPointTotalNumber = 0;
	m_dMaxRadius = 1.0;
	m_nTimeNow = -1;
	m_nClusterIdCounter = 0;
}

CMLInterStream::~CMLInterStream(void)
{
	Release();
	delete m_pDistFunc;
}

int CMLInterStream::Release(void)
{
	int i;
	for (i=0; i<(int)m_vpClusters.size(); i++)
	{
		if (m_vpClusters[i] != NULL)
		{
			delete m_vpClusters[i];
			m_vpClusters[i] = NULL;
		}
	}
	m_vpClusters.clear();
	return 0;
}

int CMLInterStream::SetParameters(const std::wstring& wstrParams)
{
	return 0;
}

int CMLInterStream::Insert(const CMLDataPoint& dataPoint, int nTime)
{
	int i;
	int nRet;
	if (m_nDataPointTotalNumber < m_nInitNumber)
	{
		m_DataSet.Insert(dataPoint);
		if (m_nDataPointTotalNumber == m_nInitNumber -1)
		{
			m_nTimeNow = nTime;
			// Start to create the Tree
			nRet = CreateClusters();
			// Empty dataset
			m_DataSet.Clear();
		}
		else
		{
			nRet = 0;
		}
	}
	else
	{
		// Find the nearest cluster
		int nNearestClusterIndex = FindNearestCluster(dataPoint);
		if (nNearestClusterIndex == -1)
		{
			m_DataSet.Insert(dataPoint);
			if (m_DataSet.Count() >= 2)
			{
				GrowNewCluster();
				// Empty dataset
				m_DataSet.Clear();
			}
		}
		else
		{
			// Cluster Absorb
			CMLInterStreamCluster* pCluster = m_vpClusters[nNearestClusterIndex];

			pCluster->Absorb(dataPoint, nTime);
			/*if (pCluster->CanAbsorb(dataPoint, m_dMaxRadius))
			{
				pCluster->Absorb(dataPoint, nTime);
			}
			else
			{
				CMLInterStreamCluster* pNewCluster = new CMLInterStreamCluster(this, m_nClusterIdCounter++);
				pNewCluster->Create(dataPoint, nTime);
				m_vpClusters.push_back(pNewCluster);
			}*/

			// Pop out outdated data point
			for (i=0; i<(int)m_vpClusters.size(); i++)
				m_vpClusters[i]->Popout(nTime - m_nTimeHorizon);
			m_nTimeNow = nTime;
		}
		nRet = 0;
	}
	m_nDataPointTotalNumber++;
	return nRet;
}

int CMLInterStream::CreateClusters(void)
{
	int i;
	CMLClustering_KMeans kMeans;
	int nK = m_nNumInitClusters;

	// Initialize the k-means clustering method
	kMeans.SetDistanceFunc(m_pDistFunc);
	kMeans.SetK(nK);

	// Use K-Means to create micro-clusters
	std::vector<CMLDataCluster> vClusters;
	int nRet = kMeans.DoClustering(m_DataSet, vClusters);
	if (ML_FAILED(nRet))
		return nRet;

	// Create the micro-clusters
	m_vpClusters.clear();
	m_vpClusters.resize((size_t)nK);
	double dAvgRadius = 0;
	for (i=0; i<nK; i++)
	{
		m_vpClusters[i] = new CMLInterStreamCluster(this, m_nClusterIdCounter++);
		m_vpClusters[i]->Create(vClusters[i].dataset);
		dAvgRadius += m_vpClusters[i]->GetRadius();
	}

	return 0;
}

int CMLInterStream::FindNearestCluster(const CMLDataPoint& point)
{
	if (m_vpClusters.size() == 0)
		return -1;

	double dMinDist = DBL_MAX;
	int nNearestIndex = -1;
	// Find the minimum distance child node
	int i;
	for (i=0; i<(int)m_vpClusters.size(); i++)
	{
		CMLInterStreamCluster* pCluster = m_vpClusters[i];
		double dDist = m_pDistFunc->Distance(point, pCluster->GetCenter());
		if (dDist < dMinDist)
		{
			nNearestIndex = i;
			dMinDist = dDist;
		}
	}

	return nNearestIndex;
}

int CMLInterStream::GrowNewCluster(void)
{
	CMLInterStreamCluster* pCluster = new CMLInterStreamCluster(this, m_nClusterIdCounter++);
	pCluster->Create(m_DataSet);
	m_vpClusters.push_back(pCluster);
	return 0;
}

int CMLInterStream::MaintainClusters(void)
{
	int nRet = 0;
	int nClusterIndex;

	int nNumClusters = (int)m_vpClusters.size();

	std::vector<CMLInterStreamCluster*> vpNewClusters;
	for (nClusterIndex=0; nClusterIndex<(int)m_vpClusters.size(); nClusterIndex++)
	{
		CMLInterStreamCluster* pCluster = m_vpClusters[nClusterIndex];
		if (pCluster == NULL)
			continue;
		if (pCluster->TriggerDieOutEvent())		// Die out 
		{
			nRet = pCluster->ExecuteDieOut();
			ML_ON_FAILED_RETURN(nRet);

			delete pCluster;
			m_vpClusters[nClusterIndex] = NULL;
			nNumClusters--;
		}
		else if (pCluster->TriggerSplitEvent())		// Max R, Split
		{
			if (nNumClusters == m_nNumMaxClusters)
			{
				// Throw out a part of the cluster
				nRet = pCluster->ExecuteThrowOut();
				ML_ON_FAILED_RETURN(nRet);
			}
			else
			{
				CMLInterStreamCluster* pNewCluster = new CMLInterStreamCluster(this, m_nClusterIdCounter++);
				nRet = pCluster->ExecuteSplit(pNewCluster);
				ML_ON_FAILED_RETURN(nRet);
				vpNewClusters.push_back(pNewCluster);
				nNumClusters++;
			}
		}

	}

	for (nClusterIndex=0; nClusterIndex<(int)m_vpClusters.size(); nClusterIndex++)
	{
		if (m_vpClusters[nClusterIndex] == NULL)
		{
			m_vpClusters.erase(m_vpClusters.begin() + nClusterIndex);
			nClusterIndex--;
		}
	}

	for(nClusterIndex=0; nClusterIndex<(int)vpNewClusters.size(); nClusterIndex++)
	{
		m_vpClusters.push_back(vpNewClusters[nClusterIndex]);
	}

	// Check maximal number of clusters
	MaintainClusterCount();

	return 0;
}

int CMLInterStream::MaintainClusterCount(void)
{
	int nRet;
	int nClusterIndex;
	int j;
	// Check maximal number of clusters
	while ((int)m_vpClusters.size() > m_nNumMaxClusters)
	{
		double dMinDist = DBL_MAX;
		int nMergeClusterIndex1 = -1;
		int nMergeClusterIndex2 = -1;
		// Find the max radius cluster
		for (nClusterIndex=0; nClusterIndex<(int)m_vpClusters.size(); nClusterIndex++)
		{
			CMLDataPoint center = m_vpClusters[nClusterIndex]->GetCenter();
			for (j=nClusterIndex+1; j<(int)m_vpClusters.size(); j++)
			{
				double dDist = m_pDistFunc->Distance(center, m_vpClusters[j]->GetCenter());
				if (dDist < dMinDist)
				{
					nMergeClusterIndex1 = nClusterIndex;
					nMergeClusterIndex2 = j;
					dDist = dMinDist;
				}
			}
		}

		// Merge the two closest clusters
		if (nMergeClusterIndex1 != -1 && nMergeClusterIndex2 != -1)
		{
			CMLInterStreamCluster* pCluster = m_vpClusters[nMergeClusterIndex1];
			CMLInterStreamCluster* pNearCluster = m_vpClusters[nMergeClusterIndex2];
			nRet = pCluster->ExecuteMerge(pNearCluster);
			ML_ON_FAILED_RETURN(nRet);
			delete pNearCluster;
			m_vpClusters.erase(m_vpClusters.begin() +nMergeClusterIndex2);
		}
		else
		{
			break;
		}
	}
	return 0;
}

